package com.etlmaster.executor.steps.scheduler;

import java.util.HashMap;
import java.util.Map;

import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.springframework.web.client.RestTemplate;

import com.etlmaster.executor.Step;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.annotation.StepInstruction;
import com.etlmaster.executor.scheduler.Proc;
import com.etlmaster.executor.scheduler.utils.ZkUtils;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.StepState;
import com.etlmaster.jdbc.JdbcTemplate;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
@StepInstruction("submitTask2Cluster")
public class SubmitToClusterStep extends Step {

	private SubmitToClusterStepMeta meta;
	protected RestTemplate template = new RestTemplate();
	private ZooKeeper zk;
	private Watcher wh;

	public SubmitToClusterStep(StepMeta meta) {
		super(meta);
		this.meta = (SubmitToClusterStepMeta) meta;
	}

	@Override
	protected int handle() {

		try {
			wh = ZkUtils.getWatcher();
			zk = ZkUtils.getInstance(
					DataCollectionUtils.getProperties("zookeeper.connect"),
					DataCollectionUtils.getIntProperties("zktimeout"), wh);
			String mainHost=new String(zk.getData(DataCollectionUtils.getProperties("fetchMainServer"),
					true, null),"utf-8");
			String url = DataCollectionUtils.getProperties("submitToCluster");
			//String fetchMainServerSql = DataCollectionUtils.getProperties("fetchMainServer");
			//JdbcTemplate jdbc = new JdbcTemplate();
			//Map<String, Object> host = jdbc.queryForMap(fetchMainServerSql);
			//String mainHost = (String) host.get("host");
			String jobname = meta.getJobname();
			String taskid = meta.getTaskid();
			Boolean concurrencyRun = Boolean.parseBoolean(meta
					.getConcurrencyRun());
			String agentId = (String) meta.getAgentId();
			Proc proc = new Proc(jobname, taskid, concurrencyRun,
					meta.getParam(), agentId);
			proc.setParam(meta.getParam());
			Map<String, Object> data = new HashMap<String, Object>();
			data.put("host", mainHost);
			data.put("param", proc.getRunParam());
			url = meta.variableSubstitution(url, data);
			meta.log("INFO","调用服务:{}",url);
			String rtn = template.getForObject(url, String.class,
					new String[] {});
			meta.getStepState().getReturnValue().put("rtnCode", rtn);
			if(Integer.parseInt(rtn)==0)
				return StepState.RUNNING_SUCCESSED;
			else
				return StepState.RUNNING_FAILED;
		} catch (Exception e) {
			meta.log("ERROR",e.getMessage());
			return StepState.RUNNING_FAILED;
		}
	}

	@Override
	public boolean stop() {
		// TODO Auto-generated method stub
		return false;
	}

}
